package com.yuninglong.cdc_demo1;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @author: yuninglong
 */
public class CdcFromMysql {
    public static void main(String[] args) throws Exception {
        new CdcFromMysql().execute(args);
    }

    private void execute(String[] args) throws Exception {

        try {
            //TODO 1）初始化flink流处理的运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 使用MySQLSource创建数据源
// 同时指定StringDebeziumDeserializationSchema，将CDC转换为String类型输出
            SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                    .hostname("192.168.202.10").port(3301)
                    .databaseList("demo").username("root").password("testYnlcom")
                    .deserializer(new StringDebeziumDeserializationSchema()).build();

// 单并行度打印，避免输出乱序
            env.addSource(sourceFunction).print().setParallelism(1);

            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
